package org.apache.ocean.main.merge;

import java.net.URL;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.lucene.index.Term;
import org.apache.lucene.queryParser.ParseException;
import org.apache.ocean.main.IndexSchema;
import org.apache.ocean.main.IndexDocuments.IndexDocument;
import org.apache.ocean.main.client.OceanXML;
import org.apache.ocean.main.http.HttpParameters;
import org.apache.ocean.main.http.HttpTimeout;
import org.apache.ocean.main.http.MulticastHttp;
import org.apache.ocean.main.http.MulticastHttp.ExceptionEnum;
import org.apache.ocean.main.search.TermFrequencies;
import org.apache.ocean.main.search.Search.Result;
import org.apache.ocean.main.search.Search.Sort;
import org.apache.ocean.main.server.TransactionServlet.HttpSearchParameters;
import org.apache.ocean.main.util.Constants;
import org.apache.ocean.main.util.Timeout;
import org.jdom.Element;

public class MergeSearch {
	private ExecutorService threadPool;
	private IndexSchema.Category schemaCategory;
	private Map<URL, Long> snapshotMap;
	private List<URL> urls;
	private HttpParameters httpParameters;
	private TermFrequencies termFrequencies;
	private Result<IndexDocument> docIdsResult;
	private Result<IndexDocument> documentsResult;
	private HttpSearchParameters searchParameters;
	private int numFound = 0;
	
	public MergeSearch(List<URL> urls, HttpParameters httpParameters, IndexSchema.Category schemaCategory, ExecutorService threadPool) throws ParseException {
		this.urls = urls;
		this.httpParameters = httpParameters;
		this.schemaCategory = schemaCategory;
		this.threadPool = threadPool;
		searchParameters = new HttpSearchParameters(httpParameters, schemaCategory);
	}
	
	public static class ResponseSummary {
		private String type;
		private long elapsed;
		private URL url;
	}
	
	public Result<IndexDocument> search() throws Exception {
		Timeout timeout = searchParameters.getTimeout();
		String queryString = httpParameters.get("q");
		searchTermFrequencies(queryString, (int)timeout.getTimeLeft());  
		searchDocIds((int)timeout.getTimeLeft());
		return getDocuments((int)timeout.getTimeLeft());
	}
	
	private Result<IndexDocument> getDocuments(int t) throws Exception {
		Timeout timeout = new Timeout(t);
		List<IndexDocument> docIdDocuments = docIdsResult.getDocs();
		Map<URL,GetDocument> map = new HashMap<URL,GetDocument>();
		for (IndexDocument document : docIdDocuments) {
			URL url = (URL)document.get("url");
			GetDocument getDocument = map.get(url);
			if (getDocument == null) {
				Long snapshotId = (Long)document.get("snapshotid");
				getDocument = new GetDocument(url, snapshotId);
			}
			getDocument.add(document);
		}
		List<URL> urls = new ArrayList<URL>();
		for (GetDocument getDocument : map.values()) {
			URL url = getDocument.getUrl();
			HttpParameters parameters = new HttpParameters();
			parameters.add("snapshotid", getDocument.getSnapshotId());
			for (IndexDocument indexDocument : getDocument.getDocuments()) {
				Long docid = (Long)indexDocument.get(Constants.ID);
				parameters.add("docid", docid);
			}
			parameters.add("fields", httpParameters.get("fields"));
			urls.add(new URL(url.toString()+"?"+parameters.toEncodedQueryString()));
		}
		MulticastHttp multicastHttp = new MulticastHttp(threadPool, ExceptionEnum.ALL);
		HttpTimeout httpTimeout = new HttpTimeout(1000, t);
		List<OceanXML> responses = multicastHttp.execute(urls, httpTimeout);
		List<IndexDocument> documents = new ArrayList<IndexDocument>();
		for (OceanXML response : responses) {
			Long snapshotid = response.getHeaderLong("snapshotid");
  		URL serverurl = response.getHeaderURL("serverurl");
  		String serverid = response.getHeaderValue("serverid");
  		String cellid = response.getHeaderValue("cellid");
			for (Element element : response.getBodyChildren()) {
				IndexDocument indexDocument = new IndexDocument(element, schemaCategory);
				indexDocument.add(Constants.SNAPSHOTID, snapshotid);
				indexDocument.add(Constants.SERVERURL, serverurl);
				indexDocument.add(Constants.SERVERID, serverid);
				indexDocument.add(Constants.CELL, cellid);
				documents.add(indexDocument);
			}
		}
		long elapsed = timeout.getElapsed();
		documentsResult = new Result<IndexDocument>(searchParameters.getStart(), documents.toArray(new IndexDocument[0]), numFound, elapsed);
		return documentsResult;
	}
	
	private static class GetDocument {
		private URL url;
		private Long snapshotId;
		private List<IndexDocument> documents = new ArrayList<IndexDocument>();
		
		public GetDocument(URL url, Long snapshotId) {
			this.url = url;
			this.snapshotId = snapshotId;
		}
		
		public URL getUrl() {
			return url;
		}

		public Long getSnapshotId() {
			return snapshotId;
		}

		public List<IndexDocument> getDocuments() {
			return documents;
		}

		public void add(IndexDocument indexDocument) {
			documents.add(indexDocument);
		}
	}
	
	private void searchDocIds(int t) throws Exception {
		Timeout timeout = new Timeout(t);
  	int numDocs = searchParameters.getStart()+searchParameters.getNum();
  	Sort sort = searchParameters.getSort();
  	if (sort == null) {
  		sort = Sort.SCORE_SORT;
  	}
  	MulticastHttp multicastHttp = new MulticastHttp(threadPool, ExceptionEnum.ALL);
  	List<URL> docIdsUrls = new ArrayList<URL>();
  	for (Map.Entry<URL, Long> entry : snapshotMap.entrySet()) {
  		HttpParameters parameters = new HttpParameters();
  		parameters.add("num", numDocs);
    	parameters.add("category", schemaCategory.getName());
    	parameters.add("snapshotid", entry.getValue());
    	parameters.add("action", "searchdocids");
  		URL url = new URL(entry.getKey()+"?"+parameters.toEncodedQueryString());
  		docIdsUrls.add(url);
  	}
  	HttpTimeout httpTimeout = new HttpTimeout(1000, t);
  	List<OceanXML> responses = multicastHttp.execute(docIdsUrls, httpTimeout);
  	MergeSortedQueue queue = new MergeSortedQueue(sort, numDocs);
  	for (OceanXML response : responses) {
  		numFound += response.getHeaderInt("numfound");
  		Long snapshotId = response.getHeaderLong("snapshotid");
  		URL serverUrl = response.getHeaderURL("serverurl");
  		String serverid = response.getHeaderValue("serverid");
  		for (Element element : response.getBodyChildren()) {
  			IndexDocument indexDocument = new IndexDocument(element, schemaCategory);
  			indexDocument.add(Constants.SNAPSHOTID, snapshotId);
  			indexDocument.add(Constants.SERVERURL, serverUrl);
  			indexDocument.add(Constants.SERVERID, serverid);
  			indexDocument.add(Constants.SNAPSHOTID, snapshotId);
  			queue.insert(indexDocument);
  		}
  	}
  	IndexDocument[] sortedDocuments = new IndexDocument[queue.size()];
		for (int i = queue.size() - 1; i >= 0; i--)
			sortedDocuments[i] = (IndexDocument) queue.pop();
		long elapsed = timeout.getElapsed();
		docIdsResult = new Result<IndexDocument>(searchParameters.getStart(), sortedDocuments, numFound, elapsed);
	}
	
	private void searchTermFrequencies(String queryString, int t) throws Exception {
		Timeout timeout = new Timeout(t);
		HttpTimeout httpTimeout = new HttpTimeout(1000, t);
		long schemaVersion = schemaCategory.getIndexSchema().getVersion();
		String category = schemaCategory.getName();
		MulticastHttp multicastHttp = new MulticastHttp(threadPool, ExceptionEnum.ALL);
		HttpParameters termFrequenciesParameters = new HttpParameters();
		termFrequenciesParameters.add("action", "gettermfrequencies");
		termFrequenciesParameters.add("category", category);
		termFrequenciesParameters.add("q", queryString);
		termFrequenciesParameters.add("schemaversion", schemaVersion);
		List<OceanXML> termFreqResponses = multicastHttp.execute(urls, termFrequenciesParameters, httpTimeout);
		termFrequencies = new TermFrequencies(schemaCategory);
		int maxDoc = 0;
		Map<URL, Long> snapshotMap = new HashMap<URL, Long>(urls.size());
		for (OceanXML termFreqResponse : termFreqResponses) {
			URL serverUrl = termFreqResponse.getHeaderURL("serverurl");
			Element element = termFreqResponse.getFirstInBody();   
			TermFrequencies nodeTermFrequencies = new TermFrequencies(element, schemaCategory);
			for (Map.Entry<Term, AtomicInteger> entry : nodeTermFrequencies.entrySet()) {
				termFrequencies.add(entry.getKey(), entry.getValue().intValue());
			}
			maxDoc += nodeTermFrequencies.getMaxDoc();
			snapshotMap.put(serverUrl, nodeTermFrequencies.getSnapshotId());
		}
		termFrequencies.setMaxDoc(maxDoc);
	}  
}
